package mr

import (
	"fmt"
	"io/ioutil"
	"os"
	"strconv"
	"strings"
	"time"
	"unicode"
)
import "log"
import "net/rpc"
import "hash/fnv"

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

func readContent(FileName string) (res string, err error) {
	file, err := os.Open(FileName)
	if err != nil {
		log.Fatalf("Cannot open %v", file)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", file)
	}
	res = string(content)
	return
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	id := os.Getpid()
	//fmt.Printf("worker%d 启动\n", id)
	for {
		time.Sleep(time.Millisecond * 500)
		args := GetTaskArgs{WorkerId: id}
		reply := GetTaskReply{}
		var MapResult []KeyValue
		intermediate := make(map[string][]string)
		ReduceResult := make(map[string]string)
		ok := call("Coordinator.GetTask", &args, &reply)
		if ok {
			//fmt.Printf("worker %d 收到任务 %v \n", id, reply.InputFile)
			if reply.Stage == 0 {
				break
			} else if reply.Stage == 1 {
				// 完成map任务
				contents, _ := readContent(reply.InputFile)
				MapResult = mapf(reply.InputFile, contents)
			} else if reply.Stage == 2 {
				// 完成reduce任务
				contents, _ := readContent(ReduceInput + reply.InputFile)
				ff := func(r rune) bool {
					return unicode.IsSpace(r)
				}
				words := strings.FieldsFunc(contents, ff)
				for i := 0; i < len(words); i += 2 {
					key := words[i]
					value := words[i+1]
					intermediate[key] = append(intermediate[key], value)

				}
				for k, v := range intermediate {

					ReduceResult[k] = reducef(k, v)
				}
			} else {
				continue
			}
		}

		// apply commit task
		replyFinish := FinishTaskReply{}
		argsFinish := FinishTaskArgs{
			WorkerId:  id,
			InputFile: reply.InputFile,
		}
		ok = call("Coordinator.FinishTask", &argsFinish, &replyFinish)

		if ok {
			if replyFinish.Ok {
				// 如果被接受则将缓存文件写入结果文件中
				if reply.Stage == 1 {
					HashKV := make(map[int][]KeyValue)
					for _, kvpair := range MapResult {

						Hash := ihash(kvpair.Key) % reply.NReduce
						HashKV[Hash] = append(HashKV[Hash], kvpair)
					}

					for i, kvpairarr := range HashKV {
						var b strings.Builder
						for _, kvpair := range kvpairarr {
							fmt.Fprintf(&b, "%v %v\n", kvpair.Key, kvpair.Value)
						}

						outputFileName := ReduceInput + strconv.Itoa(i)
						outputFile, _ := os.OpenFile(outputFileName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
						fmt.Fprintf(outputFile, b.String())
						outputFile.Close()
					}
				} else {
					outputFileName := ReduceOutput + reply.InputFile
					outputFile, _ := os.OpenFile(outputFileName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
					for k, v := range ReduceResult {
						fmt.Fprintf(outputFile, "%v %v\n", k, v)

					}
				}
				//fmt.Println(reply.InputFile + "处理完成")

				argsCommit := CommitTaskArgs{InputFile: reply.InputFile}
				replyCommit := CommitTaskReply{}
				ok = call("Coordinator.CommitTask", &argsCommit, &replyCommit)
				if ok {
					if replyCommit.Ok {
						//fmt.Println(reply.InputFile + "提交成功")
					}
				}

			} else {
				fmt.Println(reply.InputFile + "处理超时")
			}
		} else {
			fmt.Println(reply.InputFile + "call请求提交失败")
		}

	}
	//fmt.Printf("worker%d 退出\n", id)
}

// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {
	// declare an argument structure.
	args := ExampleArgs{}
	// fill in the argument(s).
	args.X = 99
	// declare a reply structure.
	reply := ExampleReply{}
	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}

// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
